use crate::model::sdk_api_result::{
    SDKResultError, DEFAULT_CONNECTION_TIMEOUT_SECS, DEFAULT_REQUEST_TIMEOUT_SECS,
    DEFAULT_REQUEST_TIMEOUT_WS_HANDSHAKE,
};
use crate::model::sdk_req_result::ReqResult;
use crate::model::storage_config::{
    OptionalP2PRPCLocalStorageConfig, P2PRPCLocalStorageConfig, APP_INFO,
    P2PRPC_LOCAL_CONFIG_FILE_NAME,
};
use crate::utils::{aes_128_gcm, ids, preferences::Preferences, query};

use async_trait::async_trait;
use futures_channel::mpsc::UnboundedSender;

use futures_util::{future, pin_mut, StreamExt};
use serde::{Deserialize, Serialize};
use std::{
    collections::HashMap,
    fmt,
    fs::{self, File, OpenOptions},
    io::{self, prelude::*, Error, ErrorKind},
    path::PathBuf,
    pin::Pin,
    sync::{Arc, RwLock},
    time::Duration,
};

use tokio::time::sleep;
use tokio_tungstenite::client_async;
use tokio_tungstenite::tungstenite::protocol::frame::Frame;
use tokio_tungstenite::tungstenite::protocol::Message;

use url::Url;

pub fn default_string() -> String {
    "".to_string()
}
// MessageData ready to send
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct MessageData {
    pub id: String,
    pub event: String,
    pub data: String,
}

// MessageToSendData ready to send
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct SendData {
    pub target_id: String,
    pub msg: MessageData,
}

// websocket sender message
#[derive(Debug, Clone)]
pub enum WsSendMessage {
    // event message
    MessageToSend(SendData),
    // ack push messge
    MessageAckPushToSend(String),
    // disconnect message
    Disconnect,
    // add a message receiver
    AddReceiver(Vec<String>, UnboundedSender<WsReceivedMessage>),
}

// websocket received message
#[derive(Debug, Clone)]
pub enum WsReceivedMessage {
    Online(
        (
            String,
            String,
            String,
            String,
            Option<UnboundedSender<WsSendMessage>>,
        ),
    ),
    Error(String),
    Text(String),
    Binary(Vec<u8>),
    Frame(Frame),
    Ping,
    AckSendResult(WsAckSendResult),
}

impl WsReceivedMessage {
    pub fn is_match_events(&self, match_events: &Vec<String>) -> bool {
        if match_events.len() == 0 {
            return true;
        }
        use WsReceivedMessage::*;
        match &self {
            Text(txt) => match Self::txt_to_message_to_received(txt) {
                Ok(MessageToReceived { event, .. }) => match_events.contains(&event),
                Err(_e) => false,
            },
            _ => true,
        }
    }
    pub fn txt_to_message_to_received(data: &str) -> Result<MessageToReceived, anyhow::Error> {
        let data = serde_json::from_str::<MessageToReceived>(data)?;
        Ok(data)
    }

    fn trait_callback(
        &self,
        events: &Vec<String>,
        t: &Box<dyn WebSoketEventCallback>,
    ) -> Result<(), anyhow::Error> {
        use WsReceivedMessage::*;
        match self {
            Online((source_id, turn_group_address, turn_user_name, turn_pwd, sender)) => t
                .on_connect_event(
                    source_id.to_owned(),
                    turn_group_address.to_owned(),
                    turn_user_name.to_owned(),
                    turn_pwd.to_owned(),
                    sender.clone(),
                ),
            Error(e) => t.on_error_event(e.to_owned()),
            Text(txt) => {
                log::info!("trait_callback Text: {}", txt);
                let MessageToReceived {
                    target_id,
                    id,
                    event,
                    data,
                } = serde_json::from_str::<MessageToReceived>(&txt)?;
                if events.is_empty() || events.contains(&event) {
                    if "__obrr__reconnect__" == &event {
                        t.on_need_reconnect(data)
                    } else {
                        t.on_ws_event(target_id, event, data, id)
                    }
                } else {
                    Ok(())
                }
            }
            Binary(_txt) => {
                log::warn!("current not support Binary");
                Ok(())
            }
            Frame(_frame) => {
                log::warn!("current not support Frame");
                Ok(())
            }
            Ping => t.on_ping_event(),
            AckSendResult(ws_send_res) => {
                let WsAckSendResult {
                    is_send_ok,
                    send_msg_id,
                    reason,
                } = ws_send_res;

                t.on_send_result(send_msg_id.to_owned(), *is_send_ok, reason.to_owned())
            }
        }
    }
}

// websocket send result
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct WsAckSendResult {
    pub is_send_ok: bool,
    pub send_msg_id: String,
    pub reason: String,
}

// websocket message to received struct
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct MessageToReceived {
    #[serde(default = "default_string")]
    pub target_id: String,
    pub id: String,
    pub event: String,
    pub data: String,
}

impl Default for MessageToReceived {
    fn default() -> Self {
        MessageToReceived {
            target_id: "".to_owned(),
            id: "0".to_owned(),
            event: "".to_owned(),
            data: "".to_owned(),
        }
    }
}

impl MessageToReceived {
    pub fn from(msg: MessageData, target_id: String) -> Self {
        let MessageData { id, event, data } = msg;
        MessageToReceived {
            id,
            event,
            data,
            target_id,
        }
    }
}

#[async_trait]
pub trait AsyncConnectWs {
    // tokio-tungstenite websocket implementation.
    async fn async_connect_ws(
        &mut self,
        client: &reqwest::Client,
        ws_group_addr: String,
        connect_addr: String,
        unique_server_id: String,
        app_secret: String,
        nonce: String,

        turn_group_address: String,
        turn_user_name: String,
        turn_user_pwd: String,
    ) -> Result<(), anyhow::Error>;
    // websocket received message callback.
    fn on_callback_senders(
        &mut self,
        callback_senders: Arc<
            std::sync::RwLock<Vec<(Vec<String>, UnboundedSender<WsReceivedMessage>)>>,
        >,
        msg: &WsReceivedMessage,
    ) -> Result<(), anyhow::Error>;
}

// websocket event call back
pub trait WebSoketEventCallback: Sync + Send {
    fn on_ws_event(
        &self,
        from_target_id: String,
        event: String,
        data: String,
        msg_id: String,
    ) -> Result<(), anyhow::Error>;
    fn on_connect_event(
        &self,
        source_id: String,
        turn_addr: String,
        turn_usr_name: String,
        turn_usr_pwd: String,
        sender: Option<UnboundedSender<WsSendMessage>>,
    ) -> Result<(), anyhow::Error>;
    fn on_error_event(&self, msg: String) -> Result<(), anyhow::Error>;
    fn on_send_result(
        &self,
        event_id: String,
        is_ok: bool,
        reason: String,
    ) -> Result<(), anyhow::Error>;
    fn on_ping_event(&self) -> Result<(), anyhow::Error>;

    fn on_need_reconnect(&self, msg: String) -> Result<(), anyhow::Error>;
}

#[async_trait]
pub trait WebSocketInnerTrait: Sync + Send {
    // add inner callback for interested websocket event
    fn add_callback(
        &mut self,
        events: Vec<String>,
        callback: Box<dyn WebSoketEventCallback>,
    ) -> String;
    // remove inner callback for interested websocket event
    fn remove_callback(
        &mut self,
        callback_id: &str,
    ) -> Option<(Vec<String>, Box<dyn WebSoketEventCallback>)>;
    // websocket async connect.
    async fn connect(&mut self) -> Result<(), anyhow::Error>;
    // websocket is connect.
    fn is_connect(&self) -> bool;
    // websoket disconnect.
    fn disconnect(&mut self) -> bool;
    // websocket call back to application
    fn on_callback(&self, msg: WsReceivedMessage);
}

////////////////////////////////
/// implementation of async eov_link_sdk web socket
////////////////////////////////
pub struct WebSocketImp {
    // callback map
    callback_map: HashMap<String, (Vec<String>, Box<dyn WebSoketEventCallback>)>,
    has_connect: bool,
    server_host: String,
    subgroup_path: String,
    server_ak: String,
    server_secret: String,
    device_package: String,
    device_id: String,
    user_id: String,
    config_file_saved_path: String,
    is_server: bool,
}

// implementation `WebSocketImp`  `Display`。
impl fmt::Display for WebSocketImp {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        // 使用 `self.number` 来表示各个数据。
        write!(f, "has_connect={},server_host={},subgroup_path={},server_ak={},server_secret={},device_package={},device_id={},config_file_saved_path={},is_server={}", 
        self.has_connect,self.server_host,self.subgroup_path,self.server_ak,self.server_secret,self.device_package,self.device_id,self.config_file_saved_path,self.is_server)
    }
}

#[async_trait]
impl WebSocketInnerTrait for WebSocketImp {
    fn add_callback(
        &mut self,
        events: Vec<String>,
        callback: Box<dyn WebSoketEventCallback>,
    ) -> String {
        let id = ids::id_generate_xbit(6);
        self.callback_map.insert(id.to_owned(), (events, callback));
        id
    }

    fn remove_callback(
        &mut self,
        callback_id: &str,
    ) -> Option<(Vec<String>, Box<dyn WebSoketEventCallback>)> {
        self.callback_map.remove(callback_id)
    }

    // websocket async connect.
    async fn connect(&mut self) -> Result<(), anyhow::Error> {
        if self.set_connect(true) {
            let mut reconnect_times = 3;
            let mut res = Ok(());
            let mut force_refresh = false;
            while reconnect_times > 0 {
                reconnect_times -= 1;
                self.set_connect(true);
                res = self.inner_connect(force_refresh).await;
                self.set_connect(false);
                match &res {
                    Ok(_) => {}
                    Err(e) => {
                        if e.to_string().to_lowercase().contains("connection refused") {
                            // if server crash or unreachable,and client will received connection refused.__rust_force_expr!
                            log::warn!("Connection refused,retry connect!");
                            force_refresh = true;
                            continue;
                        }
                        if let Some(SDKResultError::ErrorHostNeedAlternative) =
                            e.downcast_ref::<SDKResultError>()
                        {
                            log::warn!("ErrorHostNeedAlternative.refresh connect!");
                            force_refresh = true;
                            continue;
                        }
                        self.on_callback(WsReceivedMessage::Error(format!(
                            "websocket call inner_connect res err: {}",
                            e.to_string()
                        )));
                    }
                }
                break;
            }
            res
        } else {
            let msg = "websocket_impl connect call set_connect(true) return false!";
            log::warn!("{}", msg);
            anyhow::bail!(SDKResultError::ErrorAsyncAlreadyConnected)
        }
    }

    // websocket is connect.
    fn is_connect(&self) -> bool {
        self.has_connect
    }

    // websoket disconnect.
    fn disconnect(&mut self) -> bool {
        if self.has_connect {
            // break ws_to_callback await.
            self.has_connect = false;

            true
        } else {
            false
        }
    }
    // websocket call back to application
    fn on_callback(&self, msg: WsReceivedMessage) {
        self.callback_map
            .iter()
            .for_each(|(_id, (events, callback))| {
                if let Err(e) = msg.trait_callback(events, callback) {
                    log::warn!("trait_callback err: {:?}", e);
                }
            });
    }
}

// implementation of websocket
impl WebSocketImp {
    pub fn new(
        server_host: String,
        subgroup_path: String,
        server_ak: String,
        server_secret: String,
        device_package: String,
        device_id: String,
        user_id: String,
        config_file_saved_path: String,
        is_server: bool,
    ) -> Result<Self, anyhow::Error> {
        Ok(WebSocketImp {
            callback_map: HashMap::new(),
            has_connect: false,
            server_host: if server_host == "" {
                "http://127.0.0.1:19001".to_owned()
            } else {
                server_host
            },
            subgroup_path: if subgroup_path == "" {
                "/ws/get_lts_addrs".to_owned()
            } else {
                subgroup_path
            },
            server_ak: if server_ak == "" {
                "12345678901234567890123456789012".to_owned()
            } else {
                server_ak
            },
            server_secret: if server_secret == "" {
                "d2969e51fe8ba0d4415270e8eb27ed62".to_owned()
            } else {
                server_secret
            },
            device_package: if device_package == "" {
                "com.localhost.test".to_owned()
            } else {
                device_package
            },
            device_id: if device_id == "" {
                log::error!("device_id can not empty,default.device set!");
                format!("{}", "default.device")
            } else {
                device_id
            },
            user_id: if user_id == "" {
                format!("{}", "default.user")
            } else {
                user_id
            },
            config_file_saved_path: if config_file_saved_path == "" {
                "./".to_owned()
            } else {
                let config_dir_gen_path = config_file_saved_path.clone();
                let path_exit = match File::open(config_dir_gen_path.clone()) {
                    Ok(_) => true,
                    Err(err) => {
                        if std::io::ErrorKind::AlreadyExists == err.kind() {
                            true
                        } else {
                            false
                        }
                    }
                };
                if !path_exit {
                    fs::create_dir_all(config_dir_gen_path.clone()).unwrap_or_else(|why| {
                        log::error!(
                            "websocket config path {:?} mkdir failed {:?}",
                            config_dir_gen_path,
                            why.kind()
                        );
                    });
                }
                config_dir_gen_path
            },
            is_server: is_server,
        })
    }

    pub fn set_connect(&mut self, connect: bool) -> bool {
        if self.has_connect != connect {
            self.has_connect = connect;
            true
        } else {
            false
        }
    }

    pub async fn inner_connect(&mut self, force_refresh: bool) -> Result<(), anyhow::Error> {
        let device_id = &self.device_id;
        let device_pkg = &self.device_package;
        let subgroup_path = &self.subgroup_path;
        let app_key = &self.server_ak;
        let app_secret = self.server_secret.clone();
        let user_id = &self.user_id;
        let lts_host = &self.server_host;

        let chief_host: String = if self.server_host.starts_with("http://") {
            self.server_host[7..].to_owned()
        } else if self.server_host.starts_with("https://") {
            self.server_host[8..].to_owned()
        } else {
            self.server_host.to_owned()
        };
        let client = reqwest::Client::builder()
            .connect_timeout(Duration::from_secs(DEFAULT_CONNECTION_TIMEOUT_SECS))
            .gzip(true)
            // .trust_dns(true)
            .build()?;
        let config_store_path: PathBuf = self.config_file_saved_path.clone().into();
        let mut config = OptionalP2PRPCLocalStorageConfig::load(
            &APP_INFO,
            P2PRPC_LOCAL_CONFIG_FILE_NAME,
            &config_store_path,
        )
        .map_err(|e| {
            log::warn!("load config err: {}", e.to_string());
            e
        })
        .unwrap_or(OptionalP2PRPCLocalStorageConfig::default());
        log::info!("load config {:?}", config);

        let mut need_save_config = false;
        let group_lts_host_map = config.group_lts_host_map.get_or_insert_default();
        let group_turn_host_map = config.group_turn_host_map.get_or_insert_default();
        let group_nonce_map = config.group_nonce_map.get_or_insert_default();
        let group_unique_client_id_map = config.group_unique_clinet_id_map.get_or_insert_default();
        let group_turn_user_map = config.group_turn_user_map.get_or_insert_default();
        let group_turn_pwd_map = config.group_turn_pwd_map.get_or_insert_default();

        let saved_device_host_key =
            format!("{}-{}-{}-{}", chief_host, device_id, device_pkg, user_id);
        let lts_group_str = group_lts_host_map.get(&saved_device_host_key);
        let turn_group_str = group_turn_host_map.get(&saved_device_host_key);
        let nonce_str = group_nonce_map.get(&saved_device_host_key);
        let client_id_str = group_unique_client_id_map.get(&saved_device_host_key);
        let turn_user_str = group_turn_user_map.get(&saved_device_host_key);
        let turn_pwd_str = group_turn_pwd_map.get(&saved_device_host_key);
        let group_addr_struct = if force_refresh || lts_group_str.is_none() {
            let group_data = get_group_addr(
                &client,
                &lts_host,
                &subgroup_path,
                &app_key,
                &app_secret,
                &device_id,
                &device_pkg,
                &user_id,
            )
            .await?;

            group_lts_host_map.insert(
                saved_device_host_key.to_owned(),
                group_data.lts_group_address.to_owned(),
            );
            group_turn_host_map.insert(
                saved_device_host_key.to_owned(),
                group_data.turn_group_address.to_owned(),
            );
            group_nonce_map.insert(
                saved_device_host_key.to_owned(),
                group_data.nonce.to_owned(),
            );
            group_unique_client_id_map.insert(
                saved_device_host_key.to_owned(),
                group_data.unique_client_id.to_owned(),
            );
            group_turn_user_map.insert(
                saved_device_host_key.to_owned(),
                group_data.turn_user_name.to_owned(),
            );
            group_turn_pwd_map.insert(
                saved_device_host_key.to_owned(),
                group_data.turn_user_pwd.to_owned(),
            );
            need_save_config = true;
            log::info!("use new group host {}", group_data.lts_group_address);
            group_data
        } else {
            if let (
                Some(lts_host),
                Some(turn_host),
                Some(nonce),
                Some(client_id),
                Some(turn_user),
                Some(turn_pwd),
            ) = (
                lts_group_str,
                turn_group_str,
                nonce_str,
                client_id_str,
                turn_user_str,
                turn_pwd_str,
            ) {
                GroupAddrData {
                    lts_group_address: lts_host.to_owned(),
                    turn_group_address: turn_host.to_owned(),
                    nonce: nonce.to_owned(),
                    unique_client_id: client_id.to_owned(),
                    turn_user_name: turn_user.to_owned(),
                    turn_user_pwd: turn_pwd.to_owned(),
                }
            } else {
                GroupAddrData {
                    lts_group_address: "".to_owned(),
                    turn_group_address: "".to_owned(),
                    nonce: "".to_owned(),
                    unique_client_id: "".to_owned(),
                    turn_user_name: "".to_owned(),
                    turn_user_pwd: "".to_owned(),
                }
            }
        };

        if group_addr_struct.unique_client_id == "" {
            if let (
                Some(mut _group_lts_host_map),
                Some(mut _group_turn_host_map),
                Some(mut _group_nonce_map),
                Some(mut _group_uniqueid_map),
                Some(mut _group_turn_user_map),
                Some(mut _group_turn_pwd_map),
            ) = (
                config.group_lts_host_map,
                config.group_turn_host_map,
                config.group_nonce_map,
                config.group_unique_clinet_id_map,
                config.group_turn_user_map,
                config.group_turn_pwd_map,
            ) {
                _group_lts_host_map.remove(&saved_device_host_key);
                _group_turn_host_map.remove(&saved_device_host_key);
                _group_nonce_map.remove(&saved_device_host_key);
                _group_uniqueid_map.remove(&saved_device_host_key);
                _group_turn_user_map.remove(&saved_device_host_key);
                _group_turn_pwd_map.remove(&saved_device_host_key);
            }
            log::error!("group_addr_struct.unique_client_id == null.");
            anyhow::bail!(SDKResultError::ErrorHostNeedAlternative);
        } else {
            if need_save_config {
                if let Err(e) =
                    config.save(&APP_INFO, P2PRPC_LOCAL_CONFIG_FILE_NAME, &config_store_path)
                {
                    log::warn!("save config err: {:?}", e);
                } else {
                    log::info!("save config Ok {:?}", &config);
                }
            }

            let secret_bytes = hex::decode(&app_secret)?;
            let nonce_bytes = hex::decode(&group_addr_struct.nonce)?;
            let some_unique_srv_id = aes_128_gcm::aes_128_encrypt(
                &secret_bytes,
                &nonce_bytes,
                group_addr_struct.unique_client_id.to_owned(),
            );

            let mut ws_query: Vec<(String, String)> = vec![(
                "client_id".to_owned(),
                some_unique_srv_id
                    .unwrap_or(group_addr_struct.unique_client_id.to_owned())
                    .to_owned(),
            )];
            let ws_query_str =
                query::assemble_query_config((&app_key, &app_secret), &mut ws_query)?;
            let connect_addr: String = "ws://".to_owned()
                + &group_addr_struct.lts_group_address
                + "/ws/endpoint?"
                + &ws_query_str;
            log::info!("websocket connect_addr: {} ", connect_addr);

            self.async_connect_ws(
                &client,
                group_addr_struct.lts_group_address,
                connect_addr,
                group_addr_struct.unique_client_id,
                app_secret.to_owned(),
                group_addr_struct.nonce,
                group_addr_struct.turn_group_address,
                group_addr_struct.turn_user_name,
                group_addr_struct.turn_user_pwd,
            )
            .await?;
            Ok(())
        }
    }
}

fn split_first_white_space(s: &String) -> (&str, &str) {
    let bytes = s.as_bytes();

    for (i, &item) in bytes.iter().enumerate() {
        if item == b' ' {
            return (&s[0..i], &s[i..s.len()]);
        }
    }

    (&s[..], "")
}

fn inner_ws_send(
    app_secret: &String,
    nonce: &String,
    s_id: &String,
    t_id: String,
    msg: MessageData,
) -> Result<Message, tokio_tungstenite::tungstenite::Error> {
    let inner_msg = format!(
        "/send?s_id={}&t_id={}&m_id={}&m_ev={}&qos=0 {}",
        s_id,
        t_id,
        msg.id,
        msg.event,
        serde_json::to_string(&msg).unwrap_or_default()
    );
    let secret_bytes = hex::decode(&app_secret).unwrap();
    let nonce_bytes = hex::decode(&nonce).unwrap();
    let txt_encrypt = aes_128_gcm::aes_128_encrypt(&secret_bytes, &nonce_bytes, inner_msg);
    if let Some(txt) = txt_encrypt {
        return Ok(Message::Text(txt));
    };
    Err(tokio_tungstenite::tungstenite::Error::Io(Error::new(
        ErrorKind::InvalidData,
        "encrypt error!",
    )))
}

#[async_trait]
impl AsyncConnectWs for WebSocketImp {
    async fn async_connect_ws(
        &mut self,
        client: &reqwest::Client,
        group_addr: String,
        connect_addr: String,
        unique_server_id: String,
        app_secret: String,
        nonce: String,

        turn_group_address: String,
        turn_user_name: String,
        turn_user_pwd: String,
    ) -> Result<(), anyhow::Error> {
        log::info!(
            "async_connect_ws:{} unique_server_id:{} app_secret:{} nonce:{} ",
            connect_addr,
            unique_server_id,
            app_secret,
            nonce
        );

        let handshark_url = connect_addr.clone().replacen("ws", "http", 1);
        let handshark_resp_result = client
            .get(&handshark_url)
            .timeout(Duration::from_secs(DEFAULT_REQUEST_TIMEOUT_WS_HANDSHAKE))
            .send()
            .await;
        match handshark_resp_result {
            Ok(ok_status) => {
                if reqwest::StatusCode::from_u16(406).unwrap() == ok_status.status() {
                    anyhow::bail!(SDKResultError::ErrorHostNeedAlternative);
                }
            }
            Err(err_status) => {
                if let Some(err_code) = err_status.status() {
                    if reqwest::StatusCode::from_u16(406).unwrap() == err_code {
                        anyhow::bail!(SDKResultError::ErrorHostNeedAlternative);
                    }
                }
            }
        }

        let callback_senders: Arc<RwLock<Vec<(Vec<String>, UnboundedSender<WsReceivedMessage>)>>> =
            Arc::new(RwLock::new(vec![]));
        // unbond muliproduct and signal comsumer future channel
        let (send_tx, send_rx) = futures_channel::mpsc::unbounded();
        // use ws:// protocol , not tls starem

        let ws_tcp_stream = match tokio::time::timeout(
            Duration::from_secs(DEFAULT_REQUEST_TIMEOUT_SECS),
            tokio::net::TcpStream::connect(group_addr),
        )
        .await
        {
            Ok(ok) => ok,
            Err(e) => {
                log::error!("ws_tcp_stream timeout error :{}", e);
                anyhow::bail!(SDKResultError::ErrorWsHandshakeTimeout);
            }
        }?;

        let ws_stream = match client_async(connect_addr, ws_tcp_stream).await {
            Err(e) => {
                log::error!("client_async connect error :{}", e);
                anyhow::bail!(e);
            }
            Ok((ws_stream, _)) => ws_stream,
        };

        log::info!("connect_ws WebSocket handshake has been successfully completed");
        // websoket read an write stream,status changed to conneced
        let (write, mut read) = ws_stream.split();
        let send_to_ws = send_rx
            .map(|msg| match msg {
                // --------------------------------
                // todo!()
                // WsSendMessage currently not support Binary message.
                // if implementation Binary message future;
                // binary data don't need aes_128_gcm encrypt.
                // only text messages needed!
                // -----------
                // txt content ready to send
                WsSendMessage::MessageToSend(SendData { target_id, msg }) => {
                    inner_ws_send(&app_secret, &nonce, &unique_server_id, target_id, msg)
                }
                WsSendMessage::MessageAckPushToSend(pushack) => {
                    log::info!("ack push message {}", pushack);

                    let secret_bytes = hex::decode(&app_secret).unwrap();
                    let nonce_bytes = hex::decode(&nonce).unwrap();
                    let txt_encrypt =
                        aes_128_gcm::aes_128_encrypt(&secret_bytes, &nonce_bytes, pushack);
                    if let Some(txt) = txt_encrypt {
                        return Ok(Message::Text(txt));
                    };

                    Err(tokio_tungstenite::tungstenite::Error::Io(Error::new(
                        ErrorKind::InvalidData,
                        "encrypt error!",
                    )))
                }
                // disconnect message ready to send
                WsSendMessage::Disconnect => {
                    //  write message close
                    log::info!("client call close!");
                    Ok(Message::Close(None))
                }
                // new tunple with events and sender
                WsSendMessage::AddReceiver(events, receiver_sender) => {
                    match callback_senders.write() {
                        Ok(mut senders) => {
                            senders.push((events, receiver_sender));
                        }
                        Err(e) => {
                            log::error!("AddReceiver write error: {}", e.to_string())
                        }
                    }
                    Ok(Message::Pong(vec![]))
                }
            })
            .forward(write);

        // create a recieve message with mpsc::send_tx.
        let msg = WsReceivedMessage::Online((
            unique_server_id.to_owned(),
            turn_group_address.to_owned(),
            turn_user_name.to_owned(),
            turn_user_pwd.to_owned(),
            Some(send_tx.clone()),
        ));
        //  call back to sender, tunnel can received intrested event from here.
        if let Err(e) = self.on_callback_senders(callback_senders.clone(), &msg) {
            log::error!("on_sync_callback_senders err: {}", &e);
        }
        // trait_callback message.
        self.on_callback(msg);
        let duration_sec = 20;
        // create a new mpsc future channel,used for reset ticket!
        let (reset_tx, mut reset_rx) = futures_channel::mpsc::unbounded();

        // reset async function
        let reset_rec = async {
            // 20 seconds reset count
            let mut dead_delay = sleep(Duration::from_secs(duration_sec));
            // pin sleep async future.
            let mut dead_delay_pin = unsafe { Pin::new_unchecked(&mut dead_delay) };
            // if this delay future is not elapsed
            while !dead_delay_pin.is_elapsed() {
                // next async function
                // equal to futures_util::stream::Next
                let next = reset_rx.next();
                pin_mut!(next);

                // poll dead_delay_pin and poll next stream.
                match future::select(dead_delay_pin, next).await {
                    future::Either::Left(_) => {
                        // if timeout reached ,ignore and break while,
                        // then reset_rec will terminal.
                        break;
                    }
                    future::Either::Right((next_val, _)) => {
                        // if reset_rx is reached, (_,_) representation value of future1.
                        // if next value is none, then break.
                        // Poll<Option<T>>
                        if next_val.is_none() {
                            break;
                        }
                        // else reset dead_delay fucntion.
                        dead_delay = sleep(Duration::from_secs(duration_sec));
                        dead_delay_pin = unsafe { Pin::new_unchecked(&mut dead_delay) };
                    }
                }
            }
        };

        let ws_to_callback = async {
            // while connector is set
            while self.is_connect() {
                // read from websocket,read is a future split stream.
                if let Some(message) = read.next().await {
                    log::debug!("onMessage: {:?}", &message);
                    let msg = match message {
                        Ok(Message::Binary(data)) => {
                            if let Err(e) = reset_tx.unbounded_send(true) {
                                log::error!("reset_tx unbounded_send error: {}", e.to_string())
                            }
                            // dead_delay.reset(Instant::now() + Duration::from_secs(duration_sec));
                            Some(WsReceivedMessage::Binary(data))
                        }
                        Ok(Message::Frame(data)) => {
                            if let Err(e) = reset_tx.unbounded_send(true) {
                                log::error!("reset_tx unbounded_send error: {}", e.to_string())
                            }
                            // dead_delay.reset(Instant::now() + Duration::from_secs(duration_sec));
                            Some(WsReceivedMessage::Frame(data))
                        }
                        Ok(Message::Text(data)) => {
                            if let Err(e) = reset_tx.unbounded_send(true) {
                                log::error!("reset_tx unbounded_send error: {}", e.to_string())
                            }
                            let secret_bytes = hex::decode(&app_secret).unwrap();
                            let nonce_bytes = hex::decode(&nonce).unwrap();
                            let plain_text_opt =
                                aes_128_gcm::aes_128_decrypt(&secret_bytes, &nonce_bytes, data);

                            let plain_text = if let Some(text) = plain_text_opt {
                                text
                            } else {
                                log::error!("aes_128_gcm decryptor.aes_128_gcm err!");
                                "".to_owned()
                            };
                            let (illegal, content) = split_first_white_space(&plain_text);
                            let url = format!("http://localhost{}", illegal);
                            log::info!("illegal url = {},content = {}", url, content);
                            let issue_list_url = Url::parse(url.as_str());
                            let mut inner_msg = None;
                            if let Ok(url_item) = issue_list_url {
                                let vec_opt =
                                    url_item.path_segments().map(|c| c.collect::<Vec<_>>());
                                if let Some(vec) = vec_opt {
                                    if vec.len() > 0 && vec[0].starts_with("ack") {
                                        log::info!("ack received!");
                                        let hash_query: HashMap<_, _> =
                                            url_item.query_pairs().into_owned().collect();
                                        if let Some(mid) = hash_query.get("m_id") {
                                            let send_result = WsAckSendResult {
                                                is_send_ok: true,
                                                send_msg_id: mid.to_owned(),
                                                reason: "success".to_owned(),
                                            };
                                            inner_msg =
                                                Some(WsReceivedMessage::AckSendResult(send_result));
                                        } else {
                                            log::warn!("websocket recieve /ack,but mid == null,ignore,plain_text={}",plain_text);
                                        }
                                    } else if vec.len() > 0 && vec[0].starts_with("err") {
                                        log::info!("err received!");
                                        let hash_query: HashMap<_, _> =
                                            url_item.query_pairs().into_owned().collect();
                                        if let Some(mid) = hash_query.get("m_id") {
                                            let reason = if content.trim().len() > 0 {
                                                content.to_owned()
                                            } else {
                                                "unknow error".to_owned()
                                            };
                                            let send_result = WsAckSendResult {
                                                is_send_ok: false,
                                                send_msg_id: mid.to_owned(),
                                                reason: reason.to_owned(),
                                            };
                                            inner_msg =
                                                Some(WsReceivedMessage::AckSendResult(send_result));
                                        } else {
                                            log::warn!("websocket recieve /err,but mid == null,ignore,plain_text={}",plain_text);
                                        }
                                    } else if vec.len() > 0 && vec[0].starts_with("push") {
                                        let hash_query: HashMap<_, _> =
                                            url_item.query_pairs().into_owned().collect();
                                        if let (Some(mid), Some(sid)) =
                                            (hash_query.get("m_id"), hash_query.get("s_id"))
                                        {
                                            if content.trim().len() > 0 {
                                                let push_ack = format!("/ack/push?m_id={}", mid);
                                                let _ = send_tx.unbounded_send(
                                                    WsSendMessage::MessageAckPushToSend(push_ack),
                                                );
                                                let rec_msg = content.to_owned();
                                                if let Ok(mut msg_to_rec) =
                                                    serde_json::from_str::<MessageToReceived>(
                                                        &rec_msg,
                                                    )
                                                {
                                                    if msg_to_rec.target_id.len() == 0 {
                                                        msg_to_rec.target_id = sid.to_owned();
                                                    }
                                                    let with_target_msg =
                                                        serde_json::to_string(&msg_to_rec).unwrap();
                                                    inner_msg = Some(WsReceivedMessage::Text(
                                                        with_target_msg,
                                                    ));
                                                } else {
                                                    log::warn!("websocket recieve /push,but content is not json,ignore,plain_text={}",plain_text);
                                                }
                                            } else {
                                                log::warn!("websocket recieve /push,but content is null,ignore,plain_text={}",plain_text);
                                            }
                                        } else {
                                            log::warn!("websocket recieve /push,but m_id or sid is null,ignore,plain_text={}",plain_text);
                                        }
                                    } else if vec.len() > 0 && vec[0].starts_with("requery_turn") {
                                        let hash_query: HashMap<_, _> =
                                            url_item.query_pairs().into_owned().collect();
                                        if let (Some(mid), Some(reason)) =
                                            (hash_query.get("m_id"), hash_query.get("reason"))
                                        {
                                            let requery_ack =
                                                format!("/ack/requery_turn?m_id={}", mid);
                                            let _ = send_tx.unbounded_send(
                                                WsSendMessage::MessageAckPushToSend(requery_ack),
                                            );

                                            let msg_to_rec = MessageToReceived {
                                                event: "__obrr__reconnect__".to_owned(),
                                                data: reason.to_owned(),
                                                ..Default::default()
                                            };
                                            let rec_msg_json =
                                                serde_json::to_string(&msg_to_rec).unwrap();
                                            inner_msg = Some(WsReceivedMessage::Text(rec_msg_json));
                                        } else {
                                            log::warn!("websocket recieve /requery_turn,but m_id or reason == null,ignore,plain_text={}",plain_text);
                                        }
                                    }
                                };
                            };
                            inner_msg
                        }
                        Ok(Message::Close(_data)) => {
                            if !self.disconnect() {
                                let err = "Message::Close already closed".to_owned();
                                log::error!("{}", &err);
                                Some(WsReceivedMessage::Error(err))
                            } else {
                                None
                            }
                        }
                        Ok(Message::Ping(_data)) => {
                            if let Err(e) = reset_tx.unbounded_send(true) {
                                log::error!("reset_tx unbounded_send error: {}", e.to_string())
                            }
                            Some(WsReceivedMessage::Ping)
                        }
                        Ok(Message::Pong(_data)) => {
                            if let Err(e) = reset_tx.unbounded_send(true) {
                                log::error!("reset_tx unbounded_send error: {}", e.to_string())
                            }
                            Some(WsReceivedMessage::Ping)
                        }
                        Err(e) => {
                            if !self.disconnect() {
                                let err = "Message::Err already closed".to_owned();
                                log::error!("{}", &err);
                            }

                            log::error!("WS Message::Err WebSocketStream e: {}", e);
                            Some(WsReceivedMessage::Error(format!(
                                "WS Message::Err WebSocketStream err!",
                            )))
                        }
                    };
                    if let Some(msg) = msg {
                        //  send interested event to tunnel listener
                        if let Err(e) = self.on_callback_senders(callback_senders.clone(), &msg) {
                            log::error!("ws_to_callback on_sync_callback_senders err: {}", &e);
                        }
                        self.on_callback(msg);
                    }
                }
            }
        };
        // ws_stream.close(None);
        pin_mut!(send_to_ws, ws_to_callback, reset_rec);
        // let futures = vec![send_to_ws, ws_to_callback, dead_delay];
        future::select(send_to_ws, future::select(ws_to_callback, reset_rec)).await;
        return Ok(());
    }

    fn on_callback_senders(
        &mut self,
        callback_senders: Arc<
            std::sync::RwLock<Vec<(Vec<String>, UnboundedSender<WsReceivedMessage>)>>,
        >,
        msg: &WsReceivedMessage,
    ) -> Result<(), anyhow::Error> {
        match callback_senders.write() {
            Ok(mut senders) => {
                senders.drain_filter(|(events, sender)| {
                    if sender.is_closed() {
                        true
                    } else {
                        if msg.is_match_events(&events) {
                            if let Err(e) = sender.unbounded_send(msg.clone()) {
                                log::error!("callback_senders unbounded_send Err: {}", &e);
                            }
                        }
                        false
                    }
                });
            }
            Err(e) => anyhow::bail!("{}", e.to_string()),
        };
        Ok(())
    }
}

#[derive(Serialize, Deserialize)]
struct GroupAddrData {
    // websocket access server
    #[serde(rename = "lts_group_address")]
    lts_group_address: String,
    // turn group address
    #[serde(rename = "turn_group_address")]
    turn_group_address: String,
    // websocket encryption nonce, aes_gcm_128
    #[serde(rename = "nonce")]
    nonce: String,
    // when encryption is expired ,then server return 406 code
    #[serde(rename = "unique_client_id")]
    unique_client_id: String,
    // turn user name
    #[serde(rename = "turn_user_name")]
    turn_user_name: String,
    // turn user pwd
    #[serde(rename = "turn_user_pwd")]
    turn_user_pwd: String,
}

async fn get_group_addr(
    client: &reqwest::Client,
    server_host: &String,
    subgroup_path: &String,
    app_key: &String,
    secret: &String,
    device_id: &String,
    device_pkg: &String,
    user_id: &String,
) -> Result<GroupAddrData, anyhow::Error> {
    if device_id == "" || server_host == "" || device_pkg == "" {
        let msg = "device_id or host or device_package is empty!";
        log::info!("{}", msg);
        anyhow::bail!(msg);
    }
    if app_key == "" || secret == "" {
        let msg = "app_key or secret empty!";
        log::info!("{}", msg);
        anyhow::bail!(msg);
    }
    log::info!(
        "subgrouphost: {} ,device_id:{},device_pkg:{}",
        server_host,
        device_id,
        device_pkg
    );

    let mut query: Vec<(String, String)> = vec![
        ("device_id".to_owned(), device_id.to_owned()),
        ("device_pkg".to_owned(), device_pkg.to_owned()),
        ("uid".to_owned(), user_id.to_owned()),
    ];
    let query_str = query::assemble_query_config((app_key, secret), &mut query)?;

    let has_prefix = if server_host.starts_with("http://") || server_host.starts_with("https://") {
        true
    } else {
        false
    };
    let chief_url = if has_prefix {
        format!("{}{}?{}", server_host, subgroup_path, query_str)
    } else {
        format!("https://{}{}?{}", server_host, subgroup_path, query_str)
    };

    log::info!("query_url: {}", chief_url);
    // if url is not a illegal url ,then return.
    Url::parse(&chief_url)?;

    let res = client
        .get(&chief_url)
        .timeout(Duration::from_secs(DEFAULT_REQUEST_TIMEOUT_SECS))
        .send()
        .await?;
    let body = res.text().await?;
    let ReqResult { code, data, msg } = serde_json::from_str(&body)?;
    if code == 0 {
        match data {
            Some(group_data) => {
                let group_addr_data = serde_json::from_value(group_data)?;
                Ok(group_addr_data)
            }
            _ => anyhow::bail!("get group_address fail，code code = {}", code),
        }
    } else {
        if msg.is_some() {
            anyhow::bail!(
                "get group_address fail，msg.is_some() code = {},msg = {}",
                code,
                msg.unwrap()
            );
        } else {
            anyhow::bail!("get group_address fail，msg.is_none() code = {}", code);
        }
    }
}
